package saramax

import (
	"fmt"

	logger "gitee.com/youkelike/zlog"
	"gitee.com/youkelike/zmigrate/pkg/canalx"
	"github.com/IBM/sarama"
)

// 实现了接口 sarama.ConsumerGroupHandler
type CanalHandler[T any, E any] struct {
	l  logger.Logger
	fn func(msg *sarama.ConsumerMessage, t T) error
}

func NewCanalHandler[T any, E any](l logger.Logger, fn func(msg *sarama.ConsumerMessage, t T) error) *CanalHandler[T, E] {
	return &CanalHandler[T, E]{
		l:  l,
		fn: fn,
	}
}

func (h CanalHandler[T, E]) Setup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (h CanalHandler[T, E]) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (h CanalHandler[T, E]) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	msgs := claim.Messages()
	for msg := range msgs {
		h.l.Debugw("收到消息", "topic", msg.Topic)
		var t T
		h.l.Debugw("反序列化前", "消息类型", fmt.Sprintf("%T", t))
		err := canalx.UnmarshalMessage[T, E](msg.Value, &t)
		if err != nil {
			h.l.Errorw("反序列化消息失败",
				"err", err,
				"value", string(msg.Value),
				"topic", msg.Topic,
				"partition", int64(msg.Partition),
				"offset", msg.Offset)
			continue
		}
		h.l.Debugw("反序列化后", "消息内容", t)

		// 在这里执行重试
		for i := 0; i < 1; i++ {
			err = h.fn(msg, t)
			if err == nil {
				break
			}
			h.l.Errorw("处理消息失败",
				"err", err,
				"topic", msg.Topic,
				"partition", int64(msg.Partition),
				"offset", msg.Offset)
		}

		if err != nil {
			h.l.Errorw("处理消息失败-重试次数上限",
				"err", err,
				"topic", msg.Topic,
				"partition", int64(msg.Partition),
				"offset", msg.Offset)
		} else {
			session.MarkMessage(msg, "")
		}
	}
	return nil
}
